package com.atguigu.realtime.function;

import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.realtime.util.DimUtil;
import com.atguigu.realtime.util.DruidDSUtil;
import com.atguigu.realtime.util.RedisUtil;
import com.atguigu.realtime.util.ThreadPoolUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import redis.clients.jedis.Jedis;

import java.sql.SQLException;
import java.util.Collections;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @Author lzc
 * @Date 2023/3/21 14:03
 */
public abstract class DimAsyncFunction<T> extends RichAsyncFunction<T, T> {
    
    private ThreadPoolExecutor threadPool;
    
    @Override
    public void open(Configuration parameters) throws Exception {
        threadPool = ThreadPoolUtil.getThreadPool();
        
    }
    
    @Override
    public void asyncInvoke(T input, // 流中的数据, 需要对这个数据进行一步的处理
                            // 异步处理完的数据, 放入这个ResultFuture, 就会输出到后面的流中
                            ResultFuture<T> resultFuture) throws Exception {
        threadPool.submit(new Runnable() {
            @Override
            public void run() {
                // 1. 获取 redis 和 phoenix 客户端
                Jedis redisClient = RedisUtil.getRedisClient();
                DruidPooledConnection conn = DruidDSUtil.getPhoenixConn();
                // 2. 读取维度数据
                JSONObject dim = DimUtil.readDim(redisClient, conn, getTable(), getId(input));
                addDim(input, dim);
                // 3. 输出
                resultFuture.complete(Collections.singletonList(input));
                // 4. 关闭客户端(归还)
                redisClient.close();  //如果客户端是通过 new Jedis()则是关闭, 如果是通过连接池获取的则是归还
                try {
                    conn.close();
                } catch (SQLException e) {
                    throw new RuntimeException(e);
                }
            }
        });
    }
    
    public abstract void addDim(T bean, JSONObject dim) ;
    public abstract String getId(T bean) ;
    public abstract String getTable() ;
}
